package com.electric.flinkcep.function;

import com.electric.flinkcep.model.Alert;
import com.electric.flinkcep.model.Event;
import com.electric.flinkcep.service.HttpClientService;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.stream.Collectors;

/**
 * 告警处理函数
 * 用于处理CEP模式匹配的结果并生成告警
 */
public class AlertProcessFunction extends PatternProcessFunction<Event, Alert> {
    
    private static final Logger logger = LoggerFactory.getLogger(AlertProcessFunction.class);
    private final String patternName;
    private transient HttpClientService httpClientService;
    
    public AlertProcessFunction(String patternName) {
        this.patternName = patternName;
    }
    
    @Override
    public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {
        super.open(parameters);
        this.httpClientService = new HttpClientService();
        logger.info("AlertProcessFunction 初始化完成，模式名称: {}", patternName);
    }
    
    @Override
    public void processMatch(Map<String, List<Event>> match, Context ctx, Collector<Alert> out) throws Exception {
        try {
            // 收集所有匹配的事件
            List<Event> allMatchedEvents = new ArrayList<>();
            for (List<Event> events : match.values()) {
                allMatchedEvents.addAll(events);
            }
            
            if (allMatchedEvents.isEmpty()) {
                logger.warn("模式匹配结果为空，跳过处理");
                return;
            }
            
            // 按时间戳排序
            allMatchedEvents.sort(Comparator.comparingLong(Event::getTimestamp));
            
            // 生成告警
            Alert alert = createAlert(allMatchedEvents);
            
            // 输出告警到下游
            out.collect(alert);
            
            // 发送告警到后端
            sendAlertToBackend(alert);
            
            logger.info("成功处理模式匹配，生成告警: {}", alert.getAlertId());
            
        } catch (Exception e) {
            logger.error("处理模式匹配时发生错误，模式: {}, 匹配结果: {}", patternName, match, e);
            throw e;
        }
    }
    
    /**
     * 创建告警对象
     */
    private Alert createAlert(List<Event> matchedEvents) {
        Alert alert = new Alert();
        
        // 生成唯一的告警ID
        alert.setAlertId(generateAlertId());
        alert.setPatternName(patternName);
        alert.setTimestamp(System.currentTimeMillis());
        alert.setMatchedEvents(matchedEvents);
        
        // 根据模式名称设置告警类型和描述
        setAlertTypeAndDescription(alert, matchedEvents);
        
        // 设置严重级别
        alert.setSeverity(determineSeverity(matchedEvents));
        
        // 设置用户ID（如果所有事件都来自同一用户）
        String userId = extractCommonUserId(matchedEvents);
        alert.setUserId(userId);
        
        // 生成元数据
        alert.setMetadata(generateMetadata(matchedEvents));
        
        return alert;
    }
    
    /**
     * 设置告警类型和描述
     */
    private void setAlertTypeAndDescription(Alert alert, List<Event> events) {
        switch (patternName) {
            case "high-frequency-error":
                alert.setAlertType("HIGH_FREQUENCY_ERROR");
                alert.setDescription(String.format("检测到高频错误事件，在30秒内发生了%d次ERROR级别事件", events.size()));
                break;
            case "abnormal-login":
                alert.setAlertType("ABNORMAL_LOGIN");
                alert.setDescription(String.format("检测到异常登录行为，在5分钟内发生了%d次登录失败", events.size()));
                break;
            case "system-anomaly":
                alert.setAlertType("SYSTEM_ANOMALY");
                alert.setDescription("检测到系统异常，CRITICAL事件后紧跟多个ERROR事件");
                break;
            case "data-anomaly":
                alert.setAlertType("DATA_ANOMALY");
                alert.setDescription(String.format("检测到数据异常，在10秒内发生了%d次数据异常事件", events.size()));
                break;
            case "performance-degradation":
                alert.setAlertType("PERFORMANCE_DEGRADATION");
                alert.setDescription("检测到性能降级，性能警告后出现系统错误");
                break;
            case "security-threat":
                alert.setAlertType("SECURITY_THREAT");
                alert.setDescription(String.format("检测到安全威胁，在30秒内发生了%d次安全相关事件", events.size()));
                break;
            default:
                alert.setAlertType("GENERIC_ALERT");
                alert.setDescription(String.format("检测到模式匹配: %s，涉及%d个事件", patternName, events.size()));
                break;
        }
    }
    
    /**
     * 确定告警严重级别
     */
    private String determineSeverity(List<Event> events) {
        // 检查是否有CRITICAL级别的事件
        boolean hasCritical = events.stream().anyMatch(e -> "CRITICAL".equals(e.getSeverity()));
        if (hasCritical) {
            return "CRITICAL";
        }
        
        // 检查是否有ERROR级别的事件
        boolean hasError = events.stream().anyMatch(e -> "ERROR".equals(e.getSeverity()));
        if (hasError) {
            return "HIGH";
        }
        
        // 检查是否有WARNING级别的事件
        boolean hasWarning = events.stream().anyMatch(e -> "WARNING".equals(e.getSeverity()));
        if (hasWarning) {
            return "MEDIUM";
        }
        
        return "LOW";
    }
    
    /**
     * 提取共同的用户ID
     */
    private String extractCommonUserId(List<Event> events) {
        Set<String> userIds = events.stream()
                .map(Event::getUserId)
                .filter(Objects::nonNull)
                .collect(Collectors.toSet());
        
        if (userIds.size() == 1) {
            return userIds.iterator().next();
        }
        
        return null; // 多个用户或没有用户ID
    }
    
    /**
     * 生成元数据
     */
    private String generateMetadata(List<Event> events) {
        Map<String, Object> metadata = new HashMap<>();
        metadata.put("eventCount", events.size());
        metadata.put("timeSpan", events.get(events.size() - 1).getTimestamp() - events.get(0).getTimestamp());
        metadata.put("sources", events.stream().map(Event::getSource).filter(Objects::nonNull).distinct().collect(Collectors.toList()));
        metadata.put("types", events.stream().map(Event::getType).filter(Objects::nonNull).distinct().collect(Collectors.toList()));
        
        try {
            return new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(metadata);
        } catch (Exception e) {
            logger.warn("生成元数据失败", e);
            return "{}";
        }
    }
    
    /**
     * 生成唯一的告警ID
     */
    private String generateAlertId() {
        return String.format("%s-%d-%s", 
                patternName, 
                System.currentTimeMillis(), 
                UUID.randomUUID().toString().substring(0, 8));
    }
    
    /**
     * 发送告警到后端
     */
    private void sendAlertToBackend(Alert alert) {
        try {
            boolean success = httpClientService.sendAlert(alert);
            if (success) {
                logger.info("告警发送成功: {}", alert.getAlertId());
            } else {
                logger.warn("告警发送失败: {}", alert.getAlertId());
            }
        } catch (Exception e) {
            logger.error("发送告警时发生错误: {}", alert.getAlertId(), e);
        }
    }
}